package com.lzy.gmall.realtime.app.dwd;


import com.lzy.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/*
*
*                   人气店铺多维度的人气
*
* */
public class Dwd_List_of_popular_stores {
    public static void main(String[] args) throws Exception {
//      TODO 1.设施流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//      todo 2.设置并行度
        env.setParallelism(1);
//
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String topic = "dwd_traffic_page_log";
        String groupId = "dws_payment_buyer_count";

        // 添加时间生成和水位线
        tableEnv.executeSql("" +
                "create table if not exists dwd_traffic_page_log( " +
                "    `common` map<string, string>, " +//公共字段
                "    `ts` bigint, " +//设置毫秒级时间戳
                "    `rt` as TO_TIMESTAMP(FROM_UNIXTIME(ts / 1000)), " +//时间戳转为TIMESTAMP类型
                "    WATERMARK FOR rt AS rt - INTERVAL '2' SECOND " +//设置迟到两秒水位线
                " ) " + MyKafkaUtil.getKafkaDDL(topic, groupId));
//        tableEnv.executeSql("select * from dwd_traffic_page_log").print();

        // 转换行时间属性为普通时间戳
        Table pageTable = tableEnv.sqlQuery(" select " +
                " common['shop_id'] as shop_id," +
                " common['mid'] as mid," +
                " common['is_new'] as is_new," +
                " common['uid'] as user_id, " +
                "TO_TIMESTAMP(FROM_UNIXTIME(ts)) AS ts_new " +
                " from dwd_traffic_page_log "
        );
        tableEnv.createTemporaryView("page_table", pageTable);

        tableEnv.executeSql(MyKafkaUtil.getTopicDb("topic_db"));




        //      订单表
        Table order_info = tableEnv.sqlQuery(" select\n" +
                "`data`['id'] id,\n" +
                "`data`['consignee'] consignee,\n" +
                "`data`['consignee_tel'] consignee_tel,\n" +
                "`data`['total_amount'] total_amount ,\n" +
                "`data`['order_status'] order_status ,\n" +
                "CAST(`data`['user_id'] AS String)  user_id,\n" +
                "`data`['payment_way'] payment_way,\n" +
                "`data`['delivery_address'] delivery_address,\n" +
                "`data`['order_comment'] order_comment,\n" +
                "`data`['out_trade_no'] out_trade_no,\n" +
                "`data`['trade_body'] trade_body,\n" +
                "`data`['create_time'] create_time,\n" +
                "`data`['operate_time'] operate_time,\n" +
                "`data`['expire_time'] expire_time,\n" +
                "`data`['process_status'] process_status,\n" +
                "`data`['tracking_no'] tracking_no,\n" +
                "`data`['parent_order_id'] parent_order_id,\n" +
                "`data`['img_url'] img_url,\n" +
                "`data`['province_id'] province_id,\n" +
                "`data`['activity_reduce_amount'] activity_reduce_amount,\n" +
                "`data`['coupon_reduce_amount'] coupon_reduce_amount,\n" +
                "`data`['original_total_amount'] original_total_amount,\n" +
                "`data`['feight_fee'] feight_fee,\n" +
                "`data`['feight_fee_reduce'] feight_fee_reduce,\n" +
                "`data`['refundable_time'] refundable_time,\n" +
                " `type`, \n" +
                " `old` \n" +
                " from topic_db " +
                "where `database`='gmall' " +
                "and `table`='order_info' " +
                "and (`type`='insert' or `type`='update') ");
        tableEnv.createTemporaryView("order_info",order_info);




//      订单明细表
        Table order_detail = tableEnv.sqlQuery(" select\n" +
                "`data`['id'] id,\n" +
                "`data`['order_id'] order_id,\n" +
                "`data`['sku_id'] sku_id,\n" +
                "`data`['sku_name'] sku_name ,\n" +
                "`data`['img_url'] img_url ,\n" +
                "`data`['order_price'] order_price,\n" +
                "`data`['sku_num'] sku_num,\n" +
                "`data`['create_time'] create_time,\n" +
                "`data`['source_type'] source_type,\n" +
                "`data`['source_id'] source_id,\n" +
                "`data`['split_total_amount'] split_total_amount,\n" +
                "`data`['split_activity_amount'] split_activity_amount,\n" +
                "`data`['split_coupon_amount'] split_coupon_amount,\n" +
                "pt \n" +
                " from topic_db " +
                "where `database`='gmall' " +
                "and `table`='order_detail' " +
                "and `type`='insert' ");
        tableEnv.createTemporaryView("order_detail",order_detail);




//        加购表
        Table cart_info = tableEnv.sqlQuery("select " +
                "data['id'] id, " +
                "data['user_id'] user_id, " +
                "data['sku_id'] sku_id, " +
                "data['cart_price'] cart_price, " +
                "data['sku_num'] sku_num, " +
                "data['img_url'] img_url, " +
                "data['sku_name'] sku_name, " +
                "data['is_checked'] is_checked, " +
                "data['create_time'] create_time, " +
                "data['operate_time'] operate_time, " +
                "data['is_ordered'] is_ordered, " +
                "data['order_time'] order_time, " +
                "data['source_type'] source_type, " +
                "data['source_id'] source_id, " +
                "`pt` " +
                "from topic_db " +
                "where `table` = 'cart_info' " +
                "and `type` = 'update'");
        tableEnv.createTemporaryView("cart_info", cart_info);

//      收藏表
        Table favor_info = tableEnv.sqlQuery("select " +
                "data['id'] id, " +
                "data['user_id'] user_id, " +
                "data['sku_id'] sku_id, " +
                "data['spu_id'] spu_id, " +
                "data['is_cancel'] is_cancel, " +
                "data['create_time'] create_time, " +
                "data['cancel_time'] cancel_time " +
                "from topic_db " +
                "where `table` = 'favor_info' " +
                "and `type` = 'update'");
        tableEnv.createTemporaryView("favor_info", favor_info);





// 连接所有表的SQL查询
        Table joinedTable = tableEnv.sqlQuery("" +
                "SELECT " +
                "    p.shop_id, " +
                "    p.mid, " +
                "    p.user_id, " +
                "    p.is_new, " +
                "    p.ts_new, " +
                "    oi.id AS order_id, " +
                "    oi.consignee, " +
                "    oi.consignee_tel, " +
                "    oi.total_amount, " +
                "    oi.order_status, " +
                "    oi.payment_way, " +
                "    oi.delivery_address, " +
                "    oi.out_trade_no, " +
                "    oi.create_time AS order_create_time, " +
                "    od.id AS order_detail_id, " +
                "    od.sku_id, " +
                "    od.sku_name, " +
                "    od.order_price, " +
                "    od.sku_num, " +
                "    od.split_total_amount, " +
                "    od.split_activity_amount, " +
                "    od.split_coupon_amount, " +
                "    ci.id AS cart_id, " +
                "    ci.is_ordered AS is_ordered_num, " +
                "    ci.sku_num AS cart_sku_num, " +
                "    ci.create_time AS cart_create_time, " +
                "    fi.id AS favor_id, " +
                "    fi.is_cancel AS is_favor_canceled, " +
                "    fi.create_time AS favor_create_time, " +
                "    current_row_timestamp() AS row_op_ts " +
                "FROM page_table p " +
                "JOIN order_info oi ON p.user_id = oi.user_id " +
                "JOIN order_detail od ON oi.id = od.order_id " +
                "LEFT JOIN cart_info ci ON p.user_id = ci.user_id AND od.sku_id = ci.sku_id " +
                "LEFT JOIN favor_info fi ON p.user_id = fi.user_id AND od.sku_id = fi.sku_id ");

                tableEnv.createTemporaryView("joined_table", joinedTable);
//                tableEnv.toRetractStream(joinedTable, Row.class).print();


//        存储表数据到kfk
        tableEnv.executeSql("" +
                "create table dwd_page_action_order_detail( " +
                "    shop_id string, " +
                "    mid string, " +
                "    user_id string, " +
                "    is_new string, " +
                "    ts_new TIMESTAMP(3), " +
                "    order_id string, " +
                "    consignee string, " +
                "    consignee_tel string, " +
                "    total_amount string, " +
                "    order_status string, " +
                "    payment_way string, " +
                "    delivery_address string, " +
                "    out_trade_no string, " +
                "    order_create_time string, " +
                "    order_detail_id string, " +
                "    sku_id string, " +
                "    sku_name string, " +
                "    order_price string, " +
                "    sku_num string, " +
                "    split_total_amount string, " +
                "    split_activity_amount string, " +
                "    split_coupon_amount string, " +
                "    cart_id string, " +
                "    is_ordered_num string, " +
                "    cart_sku_num string, " +
                "    cart_create_time string, " +
                "    favor_id string, " +
                "    is_favor_canceled string, " +
                "    favor_create_time string, " +
                "    row_op_ts timestamp_ltz(3), " +
                "    primary key(order_detail_id) not enforced " +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_page_action_order_detail"));

        //TODO 7.将数据写出
        tableEnv.executeSql("" +
                "insert into dwd_page_action_order_detail select * from joined_table");

        // 打印测试
        tableEnv.executeSql("select * from dwd_page_action_order_detail").print();




        env.execute();
    }
}
